package cn.matezk.things.broker.hook;

import cn.hutool.core.util.NumberUtil;
import cn.matezk.things.broker.service.MqttAuthService;
import io.grpc.stub.StreamObserver;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.security.InvalidKeyException;
import java.security.NoSuchAlgorithmException;
import java.util.Arrays;
import java.util.Date;
import java.util.UUID;

@Component
@Slf4j
public class HookProvider extends io.emqx.exhook.HookProviderGrpc.HookProviderImplBase {

    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private MqttAuthService mqttAuthService;
    @Value("${mqtt.hook.auth}")
    private Boolean auth;

    @Override
    public void onProviderLoaded(io.emqx.exhook.ProviderLoadedRequest request, StreamObserver<io.emqx.exhook.LoadedResponse> responseObserver) {
        log.info("onProviderLoaded");

        io.emqx.exhook.HookSpec[] specs = {
                io.emqx.exhook.HookSpec.newBuilder().setName("client.connected").build(),
                io.emqx.exhook.HookSpec.newBuilder().setName("client.disconnected").build(),
                io.emqx.exhook.HookSpec.newBuilder().setName("client.authenticate").build(),
                io.emqx.exhook.HookSpec.newBuilder().setName("client.authorize").build(),
                io.emqx.exhook.HookSpec.newBuilder().setName("client.subscribe").build(),
                io.emqx.exhook.HookSpec.newBuilder().setName("message.publish")
                        .addTopics("things/+/property/+/+")
                        .addTopics("things/+/event/+/+")
                        .addTopics("things/+/action/+/+")
                        .addTopics("things/+/ota/+/+")
                        .build()
        };
        io.emqx.exhook.LoadedResponse reply = io.emqx.exhook.LoadedResponse.newBuilder().addAllHooks(Arrays.asList(specs)).build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }

    @Override
    public void onProviderUnloaded(io.emqx.exhook.ProviderUnloadedRequest request, StreamObserver<io.emqx.exhook.EmptySuccess> responseObserver) {
        log.info("onProviderUnloaded");
        io.emqx.exhook.EmptySuccess reply = io.emqx.exhook.EmptySuccess.newBuilder().build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }

    @Override
    public void onClientAuthorize(io.emqx.exhook.ClientAuthorizeRequest request, StreamObserver<io.emqx.exhook.ValuedResponse> responseObserver) {
        log.info("onClientAuthorize", request);
        io.emqx.exhook.ValuedResponse reply = io.emqx.exhook.ValuedResponse.newBuilder()
                .setBoolResult(true)
                .setType(io.emqx.exhook.ValuedResponse.ResponsedType.STOP_AND_RETURN)
                .build();

        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }

    @Override
    public void onClientAuthenticate(io.emqx.exhook.ClientAuthenticateRequest request, StreamObserver<io.emqx.exhook.ValuedResponse> responseObserver) {
        String clientId = request.getClientinfo().getClientid();
        String userName = request.getClientinfo().getUsername();
        String password = request.getClientinfo().getPassword();
        log.info("客户端认证 clientId: {} userName: {} password: {}", clientId, userName, password);

        Boolean success = Boolean.FALSE.equals(auth) ? Boolean.TRUE : Boolean.FALSE;
        if (auth) {
            try {
                success = mqttAuthService.auth(clientId, userName, password);
            } catch (NoSuchAlgorithmException e) {
                log.error("客户端认证失败 clientId: {} userName: {} password: {}", clientId, userName, password);
            } catch (InvalidKeyException e) {
                log.error("客户端认证失败 clientId: {} userName: {} password: {}", clientId, userName, password);
            }
        }

        io.emqx.exhook.ValuedResponse reply = io.emqx.exhook.ValuedResponse.newBuilder()
                .setBoolResult(success)
                .setType(io.emqx.exhook.ValuedResponse.ResponsedType.STOP_AND_RETURN)
                .build();

        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }

    @Override
    public void onClientSubscribe(io.emqx.exhook.ClientSubscribeRequest request, StreamObserver<io.emqx.exhook.EmptySuccess> responseObserver) {
        log.info("客户端{}订阅{}", request.getClientinfo().getClientid(), request.getTopicFiltersList());

        io.emqx.exhook.EmptySuccess reply = io.emqx.exhook.EmptySuccess.newBuilder().build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }

    @Override
    public void onClientConnected(io.emqx.exhook.ClientConnectedRequest request, StreamObserver<io.emqx.exhook.EmptySuccess> responseObserver) {
        log.info("客户端[{}]上线", request.getClientinfo().getClientid());

        if (NumberUtil.isNumber(request.getClientinfo().getUsername())) {
            MessageProperties mp = new MessageProperties();
            mp.setMessageId(String.valueOf(UUID.randomUUID()));
            mp.setTimestamp(new Date());
            String productId = request.getClientinfo().getUsername();
            String deviceId = request.getClientinfo().getClientid();
            mp.setHeader("productId", productId);
            mp.setHeader("deviceName", deviceId.substring(productId.length()));
            mp.setHeader("deviceId", deviceId);
            mp.setHeader("status", "online");
            Message message = new Message("".getBytes(StandardCharsets.UTF_8), mp);
            rabbitTemplate.convertAndSend("things.topic", "connect.status", message);
        }

        io.emqx.exhook.EmptySuccess reply = io.emqx.exhook.EmptySuccess.newBuilder().build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }

    @Override
    public void onClientDisconnected(io.emqx.exhook.ClientDisconnectedRequest request, StreamObserver<io.emqx.exhook.EmptySuccess> responseObserver) {
        log.info("客户端[{}]离线", request.getClientinfo().getClientid());

        if (NumberUtil.isNumber(request.getClientinfo().getUsername())) {
            MessageProperties mp = new MessageProperties();
            mp.setMessageId(String.valueOf(UUID.randomUUID()));
            mp.setTimestamp(new Date());
            String productId = request.getClientinfo().getUsername();
            String deviceId = request.getClientinfo().getClientid();
            mp.setHeader("productId", productId);
            mp.setHeader("deviceName", deviceId.substring(productId.length()));
            mp.setHeader("deviceId", deviceId);
            mp.setHeader("status", "offline");
            Message message = new Message("".getBytes(StandardCharsets.UTF_8), mp);
            rabbitTemplate.convertAndSend("things.topic", "connect.status", message);
        }

        io.emqx.exhook.EmptySuccess reply = io.emqx.exhook.EmptySuccess.newBuilder().build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }

    @Override
    public void onMessagePublish(io.emqx.exhook.MessagePublishRequest request, StreamObserver<io.emqx.exhook.ValuedResponse> responseObserver) {
        String topic = request.getMessage().getTopic();
        String payload = request.getMessage().getPayload().toStringUtf8();
        log.info("topic: {}, payload: {}", topic, payload);

        String items[] = topic.split("/");
        String productId = items[items.length - 2];
        String deviceName = items[items.length - 1];
        MessageProperties mp = new MessageProperties();
        mp.setMessageId(String.valueOf(UUID.randomUUID()));
        mp.setTimestamp(new Date());
        mp.setHeader("productId", productId);
        mp.setHeader("deviceName", deviceName);
        mp.setHeader("deviceId", productId.concat(deviceName));
        mp.setHeader("topic", topic);
        Message message = new Message(request.getMessage().getPayload().toByteArray(), mp);
        rabbitTemplate.convertAndSend("things.topic", String.format("dev.%s.%s", items[1], items[2]), message);

        io.emqx.exhook.ValuedResponse reply = io.emqx.exhook.ValuedResponse.newBuilder().setType(io.emqx.exhook.ValuedResponse.ResponsedType.CONTINUE).setMessage(request.getMessage()).build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }

    @Override
    public void onMessageDelivered(io.emqx.exhook.MessageDeliveredRequest request, StreamObserver<io.emqx.exhook.EmptySuccess> responseObserver) {
        String topic = request.getMessage().getTopic();
        String payload = request.getMessage().getPayload().toStringUtf8();
        log.info("topic: {}, payload: {}", topic, payload);

        String items[] = topic.split("/");
        String productId = items[items.length - 2];
        String deviceName = items[items.length - 1];
        MessageProperties mp = new MessageProperties();
        mp.setMessageId(String.valueOf(UUID.randomUUID()));
        mp.setTimestamp(new Date());
        mp.setHeader("productId", productId);
        mp.setHeader("deviceName", deviceName);
        mp.setHeader("deviceId", productId.concat(deviceName));
        mp.setHeader("topic", topic);
        Message message = new Message(request.getMessage().getPayload().toByteArray(), mp);
        rabbitTemplate.convertAndSend("things.topic", String.format("dev.%s.%s", items[1], items[2]), message);

        io.emqx.exhook.EmptySuccess reply = io.emqx.exhook.EmptySuccess.newBuilder().build();
        responseObserver.onNext(reply);
        responseObserver.onCompleted();
    }
}
